package spring.cloud.tasks.api.analyzer;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.api.database.service.StatisticsDtoService;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.NameSpaceConfiguration;
import spring.cloud.tasks.api.domain.zooKeeperClusterAndNameSpace.ZooKeeperCluster;
import spring.cloud.tasks.api.service.TaskService;
import spring.cloud.tasks.api.service.ZooKeeperClusterService;
import spring.cloud.tasks.common.ThreadFactory;
import spring.cloud.tasks.common.node.TaskNodePath;
import spring.cloud.tasks.common.zookeeper.ZooKeeper;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.*;

@Setter
@Getter
@Slf4j
public class StatisticsRefreshServiceImpl implements StatisticsRefreshService {

    private static final int STAT_THREAD_NUM = 20;
    public static int REFRESH_INTERVAL_IN_MINUTE = 7;

    private Timer refreshStatisticsTimer;
    private Timer cleanAbnormalShardingCacheTimer;
    @Resource
    private StatisticsDtoService statisticsDtoService;

    @Resource
    private StatisticsPersistence statisticsPersistence;


    @Resource
    private ZooKeeperClusterService zooKeeperService;

    @Resource
    private TaskService taskService;

    private ExecutorService executorService;

    @Resource
    private DashboardService dashboardService;

    @PostConstruct
    public void init() {
        initStatExecutorService();
        startRefreshStatisticsTimer();

    }

    @PreDestroy
    public void destroy() {
        if (executorService != null) {
            executorService.shutdownNow();
        }
        if (refreshStatisticsTimer != null) {
            refreshStatisticsTimer.cancel();
        }
        if (cleanAbnormalShardingCacheTimer != null) {
            cleanAbnormalShardingCacheTimer.cancel();
        }
    }

    private void initStatExecutorService() {
        if (executorService != null) {
            executorService.shutdownNow();
        }
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(STAT_THREAD_NUM, STAT_THREAD_NUM,
                REFRESH_INTERVAL_IN_MINUTE + 1, TimeUnit.MINUTES,
                new LinkedBlockingQueue<Runnable>(), new ThreadFactory("dashboard-statistics-thread", true));
        threadPoolExecutor.allowCoreThreadTimeOut(true);
        executorService = threadPoolExecutor;
    }

    private void startRefreshStatisticsTimer() {
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run() {
                try {
                    log.info("start refresh statistics on timer");
                    Date start = new Date();
                    List<ZooKeeperCluster> zooKeeperClusterList = zooKeeperService.getZooKeeperClusterList();
                    if (zooKeeperClusterList != null) {
                        for (ZooKeeperCluster zooKeeperCluster : zooKeeperClusterList) {
                            if (zooKeeperCluster.isOffline()) {
                                log.info("zkcluster:{} is offline, skip statistics refresh.", zooKeeperCluster.getZooKeeperClusterKey());
                                continue;
                            }

                            refreshStatistics2DB(zooKeeperCluster);
                        }
                    }

                } catch (Throwable t) {
                    log.error(t.getMessage(), t);
                }
            }
        };
        refreshStatisticsTimer = new Timer("refresh-statistics-to-db-timer", true);
        refreshStatisticsTimer
                .scheduleAtFixedRate(timerTask, 1000L * 15, 1000L * 60 * REFRESH_INTERVAL_IN_MINUTE);
    }


    @Override
    public void refresh(String zooKeeperClusterKey) throws RuntimeException {
        ZooKeeperCluster zooKeeperCluster = zooKeeperService.getZooKeeperCluster(zooKeeperClusterKey);
        if (zooKeeperCluster != null) {
            refreshStatistics2DB(zooKeeperCluster);
        } else {
            throw new RuntimeException("zk cluster not found by zkClusterKey:" + zooKeeperClusterKey);
        }
    }


    private void refreshStatistics2DB(ZooKeeperCluster zooKeeperCluster) {
        log.info("start refresh statistics by zkClusterKey:{}", zooKeeperCluster.getZooKeeperClusterKey());
        Date start = new Date();
        StatisticsAnalyzerModel statisticsAnalyzerModel = initStatisticsAnalyzerModel();
        List<Callable<Boolean>> callableList = getAnalyzeCallableList(zooKeeperCluster, statisticsAnalyzerModel);
        try {
            if (callableList != null && !callableList.isEmpty()) {
                executorService.invokeAll(callableList);
            }
            statisticsPersistence.persist(zooKeeperCluster, statisticsAnalyzerModel);


        } catch (InterruptedException e) {
            log.warn("the refreshStatistics2DB thread is interrupted", e);
            Thread.currentThread().interrupt();
        }
        log.info("end refresh statistics by zkClusterKey:{}, takes {}", zooKeeperCluster.getZooKeeperClusterKey(), new Date().getTime() - start.getTime());
    }


    protected StatisticsAnalyzerModel initStatisticsAnalyzerModel() {
        StatisticsAnalyzerModel statisticsAnalyzerModel = new StatisticsAnalyzerModel();


        TaskStatisticsAnalyzer taskStatisticsAnalyzer = new TaskStatisticsAnalyzer();
        statisticsAnalyzerModel.setTaskStatisticsAnalyzer(taskStatisticsAnalyzer);
        //
        ExecutorInfoAnalyzer executorInfoAnalyzer = new ExecutorInfoAnalyzer();
        statisticsAnalyzerModel.setExecutorInfoAnalyzer(executorInfoAnalyzer);
//


        return statisticsAnalyzerModel;
    }

    private List<Callable<Boolean>> getAnalyzeCallableList(final ZooKeeperCluster zooKeeperCluster, final StatisticsAnalyzerModel statisticsAnalyzerModel) {
        List<Callable<Boolean>> callableList = new ArrayList<>();
        for (final NameSpaceConfiguration nameSpaceConfiguration : zooKeeperCluster.getNameSpaceConfigurationList()) {
            // 过滤非当前zk连接
            if (!zooKeeperCluster.getZooKeeperAddressList().equals(nameSpaceConfiguration.getZooKeeperAddressList())) {
                continue;
            }
            Callable<Boolean> callable = new Callable<Boolean>() {
                @Override
                public Boolean call() throws Exception {
                    return analyze(zooKeeperCluster, nameSpaceConfiguration, statisticsAnalyzerModel);
                }
            };
            callableList.add(callable);
        }
        return callableList;
    }

    private boolean analyze(ZooKeeperCluster zooKeeperCluster, NameSpaceConfiguration nameSpaceConfiguration, StatisticsAnalyzerModel statisticsAnalyzerModel) {
        String nameSpace = nameSpaceConfiguration.getNameSpace();
        try {
            ZooKeeper zooKeeper = zooKeeperService.connectZooKeeper(nameSpace);
            statisticsAnalyzerModel.analyzeExecutor(zooKeeper, nameSpaceConfiguration);
            List<String> taskIdList = taskService.getTaskIdList(nameSpace);
            for (String taskId : taskIdList) {
                if (!zooKeeper.checkExists(TaskNodePath.getTaskConfigurationNodePath(taskId))) {
                    continue;
                }
                try {
                    statisticsAnalyzerModel.analyzeTaskStatistics(nameSpaceConfiguration, zooKeeper, taskId);

                } catch (Exception e) {
                    log.info(String.format("analyzeStatistics nameSpace(%s) taskId(%s) error", nameSpace, taskId), e);
                }
            }

        } catch (Exception e) {
            log.info(String.format("analyzeStatistics nameSpace(%s) error", nameSpace), e);
            return false;
        }
        return true;
    }


}